package com.yao2san.simdif.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.yao2san.common.exception.BusiException;
import com.yao2san.simdif.entity.ColumnInfo;
import com.yao2san.simdif.entity.Datasource;
import com.yao2san.simdif.entity.DatasourceConnectionAttr;
import com.yao2san.simdif.entity.TableInfo;
import com.yao2san.simdif.enums.DatasourceType;
import com.yao2san.simdif.service.*;
import com.yao2san.simdif.service.synchronizer.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.util.*;

/**
 * @author wxg
 *
 * Mysql数据源同步
 */
@Service
@Slf4j
public class DefaultDatasourceSyncServiceImpl implements DatasourceSyncService {
    @Autowired
    private DatasourceManageService datasourceManageService;
    @Autowired
    private TableInfoService tableInfoService;
    @Autowired
    private ColumnInfoService columnInfoService;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void sync(Long datasourceId) {
        Datasource datasource = datasourceManageService.getDatasource(datasourceId);
        if (datasource == null) {
            throw new BusiException("数据源不存在");
        }
        String database = datasource.getDatabase();
        String type = datasource.getType();
        String url = datasource.getUrl();
        String username = datasource.getUsername();
        String password = datasource.getPassword();
        DatasourceType datasourceType = DatasourceType.of(type);
        DatasourceSynchronizer synchronizer = null;
        Long syncId = generateSyncId();

        log.info("Start sync datasource,database:{}, type:{}, url:{} ,syncId:{}", database, type, url, syncId);
        setConnectionAttr(datasource);
        try {
            switch (datasourceType) {
                case MYSQL:
                    synchronizer = new MysqlDatasourceSynchronizer(url, username, password);
                    break;
                case H2:
                    synchronizer = new H2DatasourceSynchronizer(url, username, password);
                    break;
                case PGSQL:
                    synchronizer = new PgsqlDatasourceSynchronizer(url, username, password);
                    break;
                case ORACLE:
                    synchronizer = new OracleDatasourceSynchronizer(url, username, password);
                    break;
                case TIDB:
                default:
                    throw new IllegalArgumentException("暂不支持: " + datasourceType);
            }
            List<TableInfo> tableInfos = synchronizer.getTables(database);
            for (TableInfo info : tableInfos) {
                info.setDatasourceId(datasourceId);
                info.setSyncId(syncId);
            }
            tableInfoService.saveBatch(tableInfos);

            int columnCount = 0;
            for (TableInfo tableInfo : tableInfos) {
                List<ColumnInfo> columnInfos = synchronizer.getColumns(database, tableInfo.getTableName());
                columnInfos.forEach(columnInfo -> {
                    columnInfo.setDatasourceId(datasourceId);
                    columnInfo.setSyncId(syncId);
                    columnInfo.setTableInfoId(tableInfo.getId());
                });
                columnInfoService.saveBatch(columnInfos);
                columnCount += columnInfos.size();
            }

            updateDatasource(datasourceId, syncId, tableInfos.size());

            log.info("Datasource sync success, {} tables, {} columns", tableInfos.size(), columnCount);

        } finally {
            try {
                if (synchronizer != null) {
                    synchronizer.close();
                }
            } catch (Exception e) {
                log.error("Resource close fail", e);
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public void test(String type, String url, String username, String password) {
        DatasourceType datasourceType = DatasourceType.of(type);
        DatasourceSynchronizer synchronizer = null;
        try {
            switch (datasourceType) {
                case MYSQL:
                    synchronizer = new MysqlDatasourceSynchronizer(url, username, password);
                    break;
                case H2:
                    synchronizer = new H2DatasourceSynchronizer(url, username, password);
                    break;
                case PGSQL:
                    synchronizer = new PgsqlDatasourceSynchronizer(url, username, password);
                    break;
                case ORACLE:
                    synchronizer = new OracleDatasourceSynchronizer(url, username, password);
                    break;
                case TIDB:
                default:
                    throw new IllegalArgumentException("暂不支持: " + datasourceType);
            }
        } finally {
            try {
                if (synchronizer != null) {
                    synchronizer.close();
                }
            } catch (Exception e) {
                log.error("Resource close fail", e);
                throw new RuntimeException(e);
            }
        }
    }

    private void setConnectionAttr(Datasource datasource) {
        DatasourceType datasourceType = DatasourceType.of(datasource.getType());
        String database = datasource.getDatabase();
        DatasourceConnectionAttr connectionAttr = null;
        switch (datasourceType) {
            case MYSQL:
                connectionAttr = DatasourceConnectionAttr.builder()
                        .catalog(Optional.ofNullable(datasource.getCatalog()).orElse("def"))
                        .schema(Optional.ofNullable(datasource.getSchema()).orElse(database)).build();
                break;
            case H2:
                connectionAttr = DatasourceConnectionAttr.builder()
                        .catalog(Optional.ofNullable(datasource.getCatalog()).orElse(database))
                        .schema(Optional.ofNullable(datasource.getSchema()).orElse("'PUBLIC'")).build();
                break;
            case PGSQL:
                connectionAttr = DatasourceConnectionAttr.builder()
                        .catalog(Optional.ofNullable(datasource.getCatalog()).orElse(database))
                        .schema(Optional.ofNullable(datasource.getSchema()).orElse(database)).build();
                break;
            case ORACLE:
                connectionAttr = DatasourceConnectionAttr.builder()
                        //FIXME 待修改
                        .tablespace(Optional.ofNullable(datasource.getCatalog()).orElse(database)).build();
                break;
            default:
                connectionAttr = new DatasourceConnectionAttr();
        }
        DatasourceConnectHelper.setConnectionAttr(connectionAttr);
    }

    /**
     * 获取同步批次ID。暂时先同步执行
     *
     * @return 同步批次ID
     */
    private synchronized Long generateSyncId() {
        LambdaQueryWrapper<Datasource> query = Wrappers.lambdaQuery();
        query.orderByDesc(Datasource::getSyncId);
        List<Datasource> list = datasourceManageService.list(query);
        if (CollectionUtils.isEmpty(list) || list.get(0) == null || list.get(0).getSyncId() == null) {
            return 1L;
        }
        return list.get(0).getSyncId() + 1;
    }

    private void updateDatasource(Long datasourceId, Long syncId, Integer tableCount) {
        datasourceManageService.updateById(new Datasource()
                .setId(datasourceId)
                .setSyncId(syncId)
                .setLastSyncTime(LocalDateTime.now())
                .setTableCount(tableCount));
    }

}
